/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.servicecomb.fence.resource;

import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.apache.servicecomb.fence.api.resource.PreMethodAuthService;
import org.apache.servicecomb.fence.api.resource.RocketMqMsg;
import org.apache.servicecomb.provider.rest.common.RestSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.security.access.prepost.PreAuthorize;

import java.nio.charset.StandardCharsets;
import java.util.Collections;

@RestSchema(schemaId = "PreMethodAuthEndpoint", schemaInterface = PreMethodAuthService.class)
public class PreMethodAuthEndpoint implements PreMethodAuthService {
  protected final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private RocketMqMsgLogicService rocketMqMsgLogicService;
  @Override
  @PreAuthorize("hasRole('ADMIN')")
  public String adminSayHello(String name) {
      // Endpoint address, set to the Proxy address and port list, usually xxx:8080;xxx:8081
      String endpoint = "192.168.0.137:8081";
      // The target Topic name for message sending, which needs to be created in advance.
      String topic = "TestTopic1";
      RocketMqMsg mqMsg = new RocketMqMsg();
      mqMsg.setMsgInfo(name);
      mqMsg.setTopic(topic);
      ClientServiceProvider provider = ClientServiceProvider.loadService();
      ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
      ClientConfiguration configuration = builder.build();
      // When initializing Producer, communication configuration and pre-bound Topic need to be set.
      Producer producer = null;
      try {
          producer = provider.newProducerBuilder()
                  .setTopics(topic)
                  .setClientConfiguration(configuration)
                  .build();
      }catch (Exception e){
          logger.error("error",e);

      }

      // Sending a normal message.
      Message message = provider.newMessageBuilder()
              .setTopic(topic)
              // Set the message index key, which can be used to accurately find a specific message.
              .setKeys("messageKey")
              // Set the message Tag, used by the consumer to filter messages by specified Tag.
              .setTag("messageTag")
              // Message body.
              .setBody(name.getBytes())
              .build();
      try {
          // Send the message, paying attention to the sending result and catching exceptions.
          SendReceipt sendReceipt = producer.send(message);

          logger.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
          mqMsg.setMsgId(sendReceipt.getMessageId().toString());
          //todo

      } catch (ClientException e) {
          logger.error("Failed to send message", e);
      }
      // producer.close();
    return mqMsg.getMsgId();
  }

  @Override
  @PreAuthorize("hasRole('ADMIN')")
  public String guestSayHello(String name) {
      // Endpoint address, set to the Proxy address and port list, usually xxx:8080;xxx:8081
      String endpoint = "192.168.0.137:8081";
      // The target Topic name for message sending, which needs to be created in advance.
      String topic = "TestTopic1";
      RocketMqMsg mqMsg = new RocketMqMsg();
      mqMsg.setMsgInfo(name);
      mqMsg.setTopic(topic);
      ClientServiceProvider provider = ClientServiceProvider.loadService();
      ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
      ClientConfiguration configuration = builder.build();
      // When initializing Producer, communication configuration and pre-bound Topic need to be set.
      Producer producer = null;
      try {
          producer = provider.newProducerBuilder()
                  .setTopics(topic)
                  .setClientConfiguration(configuration)
                  .build();
      }catch (Exception e){

      }

      // Sending a normal message.
      Message message = provider.newMessageBuilder()
              .setTopic(topic)
              // Set the message index key, which can be used to accurately find a specific message.
              .setKeys("messageKey")
              // Set the message Tag, used by the consumer to filter messages by specified Tag.
              .setTag("messageTag")
              // Message body.
              .setBody(name.getBytes())
              .build();
      try {
          // Send the message, paying attention to the sending result and catching exceptions.
          SendReceipt sendReceipt = producer.send(message);

          logger.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
          mqMsg.setMsgId(sendReceipt.getMessageId().toString());
          //todo

      } catch (ClientException e) {
          logger.error("Failed to send message", e);
      }
      // producer.close();
      return mqMsg.getMsgId();
  }

  @Override
  @PreAuthorize("hasAnyRole('USER','ADMIN')")
  public String sendMsg(String name) {

      // Endpoint address, set to the Proxy address and port list, usually xxx:8080;xxx:8081
      String endpoint = "192.168.0.137:8081";
      // The target Topic name for message sending, which needs to be created in advance.
      String topic = "TestTopic1";
      RocketMqMsg mqMsg = new RocketMqMsg();
      mqMsg.setMsgInfo(name);
      mqMsg.setTopic(topic);
      ClientServiceProvider provider = ClientServiceProvider.loadService();
      ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
      ClientConfiguration configuration = builder.build();
      // When initializing Producer, communication configuration and pre-bound Topic need to be set.
      Producer producer = null;
      try {
          producer = provider.newProducerBuilder()
                  .setTopics(topic)
                  .setClientConfiguration(configuration)
                  .build();
      }catch (Exception e){

      }

      // Sending a normal message.
      Message message = provider.newMessageBuilder()
              .setTopic(topic)
              // Set the message index key, which can be used to accurately find a specific message.
              .setKeys("messageKey")
              // Set the message Tag, used by the consumer to filter messages by specified Tag.
              .setTag("messageTag")
              // Message body.
              .setBody(name.getBytes())
              .build();
      try {
          // Send the message, paying attention to the sending result and catching exceptions.
          SendReceipt sendReceipt = producer.send(message);

          logger.info("Send message successfully, messageId={}", sendReceipt.getMessageId());

      } catch (ClientException e) {
          logger.error("Failed to send message", e);
      }
      // producer.close();
      return mqMsg.getMsgId();
  }

  @Override
  public String consumeMsg(String name) {
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        // Endpoint address, set to the Proxy address and port list, usually xxx:8080;xxx:8081
        String endpoints = "192.168.0.137:8081";
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
              .setEndpoints(endpoints)
              .build();
        // Subscription message filtering rule, indicating subscription to all Tag messages.
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // Specify the consumer group the consumer belongs to, Group needs to be created in advance.
        String consumerGroup = "TestGroup";
        // Specify which target Topic to subscribe to, Topic needs to be created in advance.
        String topic ="TestTopic1";

    try {
        // Initialize PushConsumer
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                // Set the consumer group.
                .setConsumerGroup(consumerGroup)
                // Set pre-bound subscription relationship.
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                // Set the message listener.
                .setMessageListener(messageView -> {
                    // Handle messages and return the consumption result.
                    System.out.println(messageView.getMessageId());
                    logger.info("Consume message successfully, messageId={}", messageView.getMessageId());

                    String str = StandardCharsets.UTF_8.decode(messageView.getBody()).toString();
                    rocketMqMsgLogicService.insert(messageView.getMessageId().toString(),
                            topic,str);
                    //info
                    return ConsumeResult.SUCCESS;
                }).build();
        Thread.sleep(Long.MAX_VALUE);
    } catch (Exception e) {
        logger.error(e.getMessage());
    }
    return "正在接受...";
  }
}
